package global

import (
	"encoding/json"
	"fmt"
	"log"
	"runtime"
	"strings"
	"sync"
	"time"

	"github.com/gogap/errors"
	"github.com/gorilla/websocket"
	"gorm.io/gorm"
)

const (

	// Time allowed to read the next pong message from the peer.
	pongWait = 60 * time.Second

	// Maximum message size allowed from peer.
	maxMessageSize = 1024 * 2

	heartbeatInterval = 30 * time.Second // 心跳间隔

	timeoutThreshold = 10 * time.Second // 超时阈值
)

var (
	messageBatch  = make([]NoiseSecData, 0, 100) // 批量大小可根据实际情况调整
	batchMutex    sync.Mutex
	messageBatch2 = make([]MeteoroLogic, 0, 100) // 批量大小可根据实际情况调整
	batchMutex2   sync.Mutex
)

type GVA_MODEL struct {
	ID        uint           `gorm:"primarykey" json:"ID"` // 主键ID
	CreatedAt time.Time      // 创建时间
	UpdatedAt time.Time      // 更新时间
	DeletedAt gorm.DeletedAt `gorm:"index" json:"-"` // 删除时间
}

type Message struct {
	MessageType int
	Payload     []byte
}

type Client struct {
	ID         string
	Conn       *websocket.Conn
	InChan     chan *Message
	OutChan    chan *Message
	CloseChan  chan byte // 关闭通知
	Subscribes map[string]interface{}
	Server     *Server
}

type CommonReq struct {
	Type       string      `json:"type"`
	FunCode    string      `json:"funCode"`
	Token      string      `json:"token"`
	Parameters interface{} `json:"parameters"`
}

type CommonHeartbeat struct {
	Type    string `json:"type"`
	FunCode string `json:"funCode"`
	Data    string `json:"data"`
}

type CommonData struct {
	Type    string       `json:"type"`
	FunCode string       `json:"funCode"`
	Data    NoiseSecData `json:"data"`
}

type Server struct {
	Clients      map[string]*Client
	ClientsMux   sync.Mutex
	Broadcast    chan *Message
	AddClient    chan *Client
	RemoveClient chan *Client
	Stop         chan struct{}
}

type NoiseSecData struct {
	ID              uint    `json:"id" gorm:"column:id;primary_key:true"`
	Class           string  `json:"Class" gorm:"column:class"`
	Flag            string  `json:"Flag" gorm:"column:flag"`
	DeviceAWAID     string  `json:"DeviceAWAID" gorm:"column:device_awaid"`
	DateTime        string  `json:"DateTime" gorm:"column:datetime"`
	Lp              string  `json:"Lp" gorm:"column:lp"`
	LAFp            float64 `json:"LAFp" gorm:"column:lafp"`
	LCSp            float64 `json:"LCSp" gorm:"column:lcsp"`
	LZIp            string  `json:"LZIp" gorm:"column:lzip"`
	LAmax           string  `json:"LAmax" gorm:"column:lamax"`
	LAmin           string  `json:"LAmin" gorm:"column:lamin"`
	LCpeak          string  `json:"LCpeak" gorm:"column:lcpeak"`
	LAeqT           string  `json:"LAeqT" gorm:"column:laeqt"`
	LAeq1s          string  `json:"LAeq1s" gorm:"column:laeq1s"`
	IsEffective     string  `json:"isEffective" gorm:"column:iseffective"`
	EffectReason    string  `json:"effectReason" gorm:"column:effect_reason"`
	Longitude       float64 `json:"LONG" gorm:"column:longitude"`
	Latitude        float64 `json:"Latitude" gorm:"column:latitude"`
	DataLibraryFlag string  `json:"DataLibraryFlag" gorm:"column:data_library_flag"`
}

func (NoiseSecData) TableName() string {
	return "hp_zhejianghuanke_noise"
}

type Device struct {
	ID            uint    `json:"id" gorm:"column:id;primary_key:true"`
	DeviceNo      string  `json:"deviceNo" gorm:"column:device_number"`
	DeviceName    string  `json:"brand" gorm:"column:device_name"`
	StationName   string  `json:"locationInfoName" gorm:"column:station_name"`
	Region        string  `json:"address" gorm:"column:region"`
	AddressDetail string  `json:"detailAddress" gorm:"column:address_detail"`
	ResourceClass string  `json:"resourceClass" gorm:"column:resource_class"`
	FunctionClass string  `json:"functionClassax" gorm:"column:function_class"`
	Lng           float64 `json:"lng" gorm:"column:lng"`
	Lat           float64 `json:"lat" gorm:"column:lat"`
	CreateTime    string  `json:"createTime" gorm:"column:createTime"`
}

func (Device) TableName() string {
	return "hp_zhejianghuanke_device"
}

type StatMeta struct {
	LNmax string `json:"Lnmax" gorm:"column:Lnmax"`
	L90   string `json:"L90" gorm:"column:l90"`
	Lmin  string `json:"Lmin" gorm:"column:lmin"`
	Ln    string `json:"Ln" gorm:"column:ln"`
	LDmin string `json:"Ldmin" gorm:"column:ldmin"`
	L50   string `json:"L50" gorm:"column:l50"`
	L5    string `json:"L5" gorm:"column:l5"`
	L95   string `json:"L95" gorm:"column:l95"`
	L10   string `json:"L10" gorm:"column:l10"`
	SD    string `json:"SD" gorm:"column:sd"`
	LDmax string `json:"Ldmax" gorm:"column:ldmax"`
	LNmin string `json:"Lnmin" gorm:"column:lnmin"`
	Ldn   string `json:"Ldn" gorm:"column:ldn"`
	Ld    string `json:"Ld" gorm:"column:ld"`
	Sel   string `json:"SEL" gorm:"column:sel"`
	Lmax  string `json:"Lmax" gorm:"column:lmax"`
	LeqT  string `json:"LeqT" gorm:"column:leqt"`
}

type NoiseStatData struct {
	ID          uint     `json:"id"`
	Class       string   `json:"Class"`
	Flag        string   `json:"Flag"`
	DeviceAWAID string   `json:"DeviceAWAID"`
	DateTime    string   `json:"DateTime"`
	Interface   string   `json:"Interface"`
	DataType    int      `json:"dataType"`
	Data        StatMeta `json:"data"`
}

type MeteoroLogic struct {
	ID                uint   `json:"id" gorm:"column:id;primary_key:true"`
	Class             string `json:"Class" gorm:"column:class"`
	Flag              string `json:"Flag" gorm:"column:flag"`
	DeviceAWAID       string `json:"DeviceAWAID" gorm:"column:device_awaid"`
	DateTime          string `json:"DateTime" gorm:"column:datetime"`
	Temperature       string `json:"A-Temp" gorm:"column:temperature"`
	Humidity          string `json:"Humi-R" gorm:"column:humidity"`
	WindSpeed         string `json:"W-Speed" gorm:"column:wind_speed"`
	WindDirection     string `json:"W-Direction" gorm:"column:wind_direction"`
	MaxSpeed          string `json:"W-maxSpeed" gorm:"column:max_speed"`
	MaxSpeedDirection string `json:"W-maxSpeedDirection" gorm:"column:max_speed_direction"`
	AvgSpeed          string `json:"W-AverageSpeed" gorm:"column:avg_speed"`
	AirPressure       string `json:"Ari-P" gorm:"column:air_pressure"`
	AccumlateRainfall string `json:"RF-All" gorm:"column:accumlate_rainfall"`
	Datatype          string `json:"dataType" gorm:"column:datatype"`
}

func (MeteoroLogic) TableName() string {
	return "hp_zhejianghuanke_mete"
}

func NewServer() *Server {
	return &Server{
		Clients:      make(map[string]*Client),
		Broadcast:    make(chan *Message),
		AddClient:    make(chan *Client),
		RemoveClient: make(chan *Client),
		Stop:         make(chan struct{}),
	}
}

// IsSubscribedTo 用于检查客户端是否订阅了某个主题。
func (c *Client) IsSubscribedTo(topic string) (bool, interface{}) {
	pack, ok := c.Subscribes[topic]
	return ok, pack
}

func (c *Client) wsRead() (*Message, error) {
	select {
	case msg := <-c.InChan:
		return msg, nil
	case <-c.CloseChan:
	}
	return nil, errors.New("websocket closed")
}

func (c *Client) wsWrite(messageType int, data []byte) error {
	select {
	case c.OutChan <- &Message{messageType, data}:
	case <-c.CloseChan:
		return errors.New("websocket closed")
	}
	return nil
}

func (c *Client) wsClose() {
	c.Conn.Close()
}

func (s *Server) wsReadLoop(client *Client) {
	defer func() {
		s.RemoveClient <- client
	}()

	client.Conn.SetReadLimit(maxMessageSize)
	client.Conn.SetReadDeadline(time.Now().Add(pongWait))
	client.Conn.SetPongHandler(func(string) error { client.Conn.SetReadDeadline(time.Now().Add(pongWait)); return nil })

	for {
		messageType, message, err := client.Conn.ReadMessage()
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
				log.Printf("error: %v", err)
			}
			break
		}
		if messageType == websocket.PongMessage {
			log.Println(client.ID, "接收到pong消息")
			continue
		}
		if messageType == websocket.TextMessage {
			// 处理接收到的消息，例如解析JSON并更新订阅状态
			client.InChan <- &Message{MessageType: messageType, Payload: message}
		} else {
			log.Printf("Unsupported message type from client %s: %d", client.ID, messageType)
		}
		time.Sleep(100 * time.Millisecond) // 延时100毫秒
	}
}

func (s *Server) wsWriteLoop(client *Client) {
	for {
		select {
		// 取一个应答
		case msg := <-client.OutChan:
			// 写给websocket
			if err := client.Conn.WriteMessage(msg.MessageType, msg.Payload); err != nil {
				goto error
			}
		case <-client.CloseChan:
			goto closed
		}
		time.Sleep(100 * time.Millisecond) // 延时100毫秒
	}
error:
	client.wsClose()
closed:
}

func (s *Server) wsProcLoop(client *Client) {
	// 启动一个gouroutine发送心跳
	go startHeartbeat(client)
	for {
		msg, err := client.wsRead()
		if err != nil {
			fmt.Println("read fail")
			break
		}
		if msg != nil {
			switch msg.MessageType {
			case websocket.TextMessage:
				// 在这里处理文本消息，先反序列化请求，再根据功能码分发请求
				var request CommonReq
				if err := json.Unmarshal(msg.Payload, &request); err != nil {
					res := CommonHeartbeat{
						Type: "response",
						Data: "request body is illegal",
					}
					resByte, _ := json.Marshal(res)
					client.wsWrite(msg.MessageType, resByte)
					continue
				}
				if request.Token != "B905FEECF4D4F5247F0D26D5B7FC5AF1" {
					res := CommonHeartbeat{
						Type: "response",
						Data: "token is illegal",
					}
					resByte, _ := json.Marshal(res)
					client.wsWrite(msg.MessageType, resByte)
					continue
				}
				if request.Type == "subscribe" && request.Parameters != nil {
					client.Subscribes[request.FunCode] = request.Parameters
				}
				if request.Type == "unsubscribe" {
					delete(client.Subscribes, request.FunCode)
				}
				if request.Type == "request" && request.FunCode == "ReportDataReq" {
					if m, ok := request.Parameters.(map[string]interface{}); ok {
						m["type"] = "update"
						m["funCode"] = "zjhk_noise_second"
						v, _ := json.Marshal(m)

						s.Broadcast <- &Message{MessageType: websocket.TextMessage, Payload: v}
					}

					v, _ := json.Marshal(request.Parameters)
					handleNoiseSecMsg(v)
					res := CommonHeartbeat{
						Type:    "response",
						FunCode: "ReportDataReq",
						Data:    "success",
					}
					resByte, _ := json.Marshal(res)
					err = client.wsWrite(msg.MessageType, resByte)
					if err != nil {
						fmt.Println("write fail")
						break
					}
				}
				if request.Type == "request" && request.FunCode == "ReportDeviceReq" {
					v, _ := json.Marshal(request.Parameters)
					err := handleDeviceMsg(v)
					var res CommonHeartbeat
					if err != nil {
						res = CommonHeartbeat{
							Type:    "response",
							FunCode: "ReportDeviceReq",
							Data:    "error",
						}
					} else {
						res = CommonHeartbeat{
							Type:    "response",
							FunCode: "ReportDeviceReq",
							Data:    "success",
						}
					}
					resByte, _ := json.Marshal(res)
					err = client.wsWrite(msg.MessageType, resByte)
					if err != nil {
						fmt.Println("write fail")
						break
					}
				}
				if request.Type == "request" && request.FunCode == "ReportMeteReq" {
					if m, ok := request.Parameters.(map[string]interface{}); ok {
						m["type"] = "update"
						m["funCode"] = "zjhk_mete_second"
						v, _ := json.Marshal(m)

						s.Broadcast <- &Message{MessageType: websocket.TextMessage, Payload: v}
					}
					v, _ := json.Marshal(request.Parameters)
					res := CommonHeartbeat{
						Type:    "response",
						FunCode: "ReportMeteReq",
						Data:    "success",
					}
					resByte, _ := json.Marshal(res)
					err = client.wsWrite(msg.MessageType, resByte)
					if err != nil {
						fmt.Println("write fail")
						break
					}
					handleMeteMsg(v)
				}
				if request.Type == "request" && request.FunCode == "ReportStatReq" {
					v, _ := json.Marshal(request.Parameters)
					err := handleStatMsg(v)
					var res CommonHeartbeat
					if err != nil {
						res = CommonHeartbeat{
							Type:    "response",
							FunCode: "ReportStatReq",
							Data:    "error",
						}
					} else {
						res = CommonHeartbeat{
							Type:    "response",
							FunCode: "ReportStatReq",
							Data:    "success",
						}
					}
					resByte, _ := json.Marshal(res)
					err = client.wsWrite(msg.MessageType, resByte)
					if err != nil {
						fmt.Println("write fail")
						break
					}
				}
				if request.Type == "heartbeat" {
					log.Println(client.ID, "接收到pong消息")
					continue
				}
			case websocket.BinaryMessage:
				// 在这里处理二进制消息
			default:
				fmt.Println("Received unknown message type")
			}
		}
		runtime.Gosched()
	}
}

func handleNoiseSecMsg(p []byte) {
	var data NoiseSecData
	err := json.Unmarshal(p, &data)
	if err != nil {
		// 处理错误
		fmt.Println("unmarshal noise data failed:", err)
		return
	}
	batchMutex.Lock()
	messageBatch = append(messageBatch, data)
	if len(messageBatch) >= 100 { // 达到批量大小，执行插入操作
		go insertBatch(messageBatch)
		messageBatch = make([]NoiseSecData, 0, 100) // 重置批量数据
	}
	batchMutex.Unlock()
}

func handleDeviceMsg(p []byte) error {
	var data Device
	err := json.Unmarshal(p, &data)
	if err != nil {
		// 处理错误
		fmt.Println("unmarshal device failed:", err)
		return err
	}
	res := GVA_DB.Create(&data)
	return res.Error
}

func handleMeteMsg(p []byte) {
	var data MeteoroLogic
	err := json.Unmarshal(p, &data)
	if err != nil {
		// 处理错误
		fmt.Println("unmarshal meteoroLogic failed:", err)
		return
	}
	batchMutex2.Lock()
	messageBatch2 = append(messageBatch2, data)
	if len(messageBatch2) >= 100 { // 达到批量大小，执行插入操作
		go insertBatchMete(messageBatch2)
		messageBatch2 = make([]MeteoroLogic, 0, 100) // 重置批量数据
	}
	batchMutex2.Unlock()
}

func handleStatMsg(p []byte) error {
	var data NoiseStatData
	err := json.Unmarshal(p, &data)
	if err != nil {
		// 处理错误
		fmt.Println("unmarshal static failed:", err)
		return err
	}
	err = GVA_DB.Exec("INSERT INTO hp_zhejianghuanke_stat (class,flag,device_awaid,datetime,interface,datatype,Lnmax,l90,lmin,ln,ldmin,l50,l5,l95,l10,sd,ldmax,lnmin,ldn,ld,sel,lmax,leqt) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", data.Class, data.Flag, data.DeviceAWAID, data.DateTime, data.Interface, data.DataType, data.Data.LNmax, data.Data.L90, data.Data.Lmin, data.Data.Ln, data.Data.LDmin, data.Data.L50, data.Data.L5, data.Data.L95, data.Data.L10, data.Data.SD, data.Data.LDmax, data.Data.LNmin, data.Data.Ldn, data.Data.Ld, data.Data.Sel, data.Data.Lmax, data.Data.LeqT).Error
	return err
}

func insertBatch(batch []NoiseSecData) {
	res := GVA_DB.Create(&batch)
	if res.Error != nil {
		fmt.Println("Error:", res.Error)
	} else {
		fmt.Println("Rows Affected:", res.RowsAffected)
	}
}

func insertBatchMete(batch []MeteoroLogic) {
	res := GVA_DB.Create(&batch)
	if res.Error != nil {
		fmt.Println("Error:", res.Error)
	} else {
		fmt.Println("Rows Affected:", res.RowsAffected)
	}
}

func startHeartbeat(client *Client) {
	ticker := time.NewTicker(heartbeatInterval)
	defer ticker.Stop()
	for {
		select {
		case <-client.CloseChan:
			return
		case <-ticker.C:
			err := client.Conn.WriteControl(websocket.PingMessage, []byte("ping"), time.Now().Add(pongWait))
			if err != nil {
				log.Printf("Failed to send heartbeat signal: %v", err)
				client.wsClose()
				return
			}
			msg := CommonHeartbeat{
				Type: "heartbeat",
				Data: "ping",
			}
			v, err := json.Marshal(msg)
			err = client.Conn.WriteMessage(websocket.TextMessage, v)
			if err != nil {
				log.Printf("Failed to send heartbeat message: %v", err)
				client.wsClose()
				return
			}
		}
	}
}

func (s *Server) Run() {
	for {
		select {
		case client := <-s.AddClient:
			s.ClientsMux.Lock()
			s.Clients[client.ID] = client
			fmt.Println(client.ID, "join")
			s.ClientsMux.Unlock()
			go s.wsProcLoop(client)
			go s.wsReadLoop(client)
			go s.wsWriteLoop(client)

		case client := <-s.RemoveClient:
			s.ClientsMux.Lock()
			delete(s.Clients, client.ID)
			fmt.Println(client.ID, "leave")
			s.ClientsMux.Unlock()
			close(client.InChan)
			client.Conn.Close()

		case msg := <-s.Broadcast:
			s.ClientsMux.Lock()
			for _, client := range s.Clients {
				if ok, p := client.IsSubscribedTo("zjhk_noise_second"); ok {
					var strMsg = string(msg.Payload)
					for _, v := range p.([]interface{}) {
						if strings.Contains(strMsg, v.(string)) && strings.Contains(strMsg, "zjhk_noise_second") {
							client.wsWrite(msg.MessageType, msg.Payload)
						}
					}
				}
				if ok, p := client.IsSubscribedTo("zjhk_mete_second"); ok {
					var strMsg = string(msg.Payload)
					for _, v := range p.([]interface{}) {
						if strings.Contains(strMsg, v.(string)) && strings.Contains(strMsg, "zjhk_mete_second") {
							client.wsWrite(msg.MessageType, msg.Payload)
						}
					}
				}
			}
			s.ClientsMux.Unlock()
		case <-s.Stop:
			s.ClientsMux.Lock()
			for _, client := range s.Clients {
				close(client.InChan)
				client.Conn.Close()
			}
			s.Clients = make(map[string]*Client)
			s.ClientsMux.Unlock()
			return
		}
	}
}
